package com.study.flink.datastream

import org.apache.flink.streaming.api.functions.co.CoMapFunction
import org.apache.flink.streaming.api.scala.{ConnectedStreams, DataStream, StreamExecutionEnvironment}

/**
  * Flink Connect KeyBy Demo
  *
  * @author stephen
  * @create 2019-05-25 14:03
  * @since 1.0.0
  */
object FlinkConnectKeyByDemo {

  def main(args: Array[String]): Unit = {

    // 1、执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2、指定数据源
    import org.apache.flink.api.scala._
    val dataStream1 = env.fromElements(("a", 1), ("b", 2), ("c", 3), ("d", 4), ("a", 5), ("b", 6))
    val dataStream2 = env.fromElements(1, 2, 3, 4, 5, 6)

    // 3、对数据进行操作
    val connectedStream: ConnectedStreams[(String, Int), Int] = dataStream1.connect(dataStream2)
    val keyConnectedStream = connectedStream.keyBy(a => a._2, b => b)
    val resultStream: DataStream[(Int, String)] = keyConnectedStream.map(new CoMapFunction[(String, Int), Int, (Int, String)] {
      // 第一个DataStream
      override def map1(value: (String, Int)): (Int, String) = {
        (value._2, value._1)
      }

      // 第二个DataStream
      override def map2(value: Int): (Int, String) = {
        (value, "default")
      }
    })

    // 4、指定数据输出位置
    resultStream.print()

    // 5、触发任务执行
    env.execute("Flink Connect KeyBy Demo")
  }
}
